package com.haizhi.messagehub.service.redis;

import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import com.haizhi.common.exception.HZException;
import com.haizhi.messagehub.bean.form.ConsumerAdapterForm;
import com.haizhi.messagehub.bean.form.MessageForm;
import com.haizhi.messagehub.service.MessageHubServiceImpl;

@Slf4j
@Service("redisStreamProcessor")
public class RedisStreamProcessor extends MessageHubServiceImpl {


    @Override
    public void producer(MessageForm messageForm) {
        ObjectRecord<String, Object> stringObjectRecord = ObjectRecord.create(messageForm.getTopic(), messageForm.getMessage());
        stringRedisTemplate.opsForStream().add(stringObjectRecord);
    }

    @Async
    @Override
    public void consumer(ConsumerAdapterForm messageForm) {

        String topic = messageForm.getTopic();
        String group = messageForm.getGroup();
        String consumerName = messageForm.getConsumerName();
        Object consumerBean = messageForm.getBean();
        Method invokeMethod = messageForm.getInvokeMethod();

        if (ObjectUtils.isEmpty(group)) {
            throw new HZException("未指定消费者组");
        }

        new Thread(() -> {

            StreamOperations<String, String, Object> streamOperations = this.stringRedisTemplate.opsForStream();
            StreamInfo.XInfoGroups groups = streamOperations.groups(topic);

            AtomicReference<Boolean> groupHasKey = new AtomicReference<>(false);

            groups.forEach(groupInfo -> {
                if (Objects.equals(group, groupInfo.getRaw().get("name"))) {
                    groupHasKey.set(true);
                }
            });
            if (groups.isEmpty() || !groupHasKey.get()) {
                String groupName = streamOperations.createGroup(topic, group);
                log.info("creatGroup:{}", groupName);
            }

            while (true) {
                try {
                    List<ObjectRecord<String, String>> messageList = streamOperations.read(
                            String.class,
                            Consumer.from(group, consumerName),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(topic, ReadOffset.lastConsumed()));

                    if (ObjectUtils.isEmpty(messageList)) {
                        // 如果为null，说明没有消息，继续下一次循环
                        Thread.sleep(1000);
                        continue;
                    }

                    ObjectRecord<String, String> record = messageList.get(0);
                    invokeMethod.invoke(consumerBean, record.getValue());

                    stringRedisTemplate.opsForStream().acknowledge(group, record);
                } catch (Exception e) {

                    log.error(e.getMessage(), e);
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }).start();
    }
}
